package com.study.flink.window

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
  * Count Window Demo
  * CountWindow根据窗口中相同key元素的数量来触发执行，
  * 执行时只计算元素数量达到窗口大小的key对应的结果。
  * @author stephen
  * @date 2019-07-22 16:18
  */
object ScalaCountWindowDemo {

  def main(args: Array[String]): Unit = {
    // 1 获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 2 获取数据
    import org.apache.flink.api.scala._
    val dataStream = env.socketTextStream("localhost", 9999)

    // 3 Transformation
    val resultStream = dataStream
      .flatMap(_.split(" "))
      .map((_, 1))
      .keyBy(_._1)
      // 滚动窗口：默认，当相同的key达到2个计算一次
      //.countWindow(2)
      // 滑动窗口：当key的个数每达到2的时候,触发计算，计算最近该key最近10个元素的内容
      .countWindow(10,2)
      .reduce((x, y) => (x._1, x._2 + y._2))

    // 4 输出
    resultStream.print()

    // 5 启动任务
    env.execute("Count Window Demo")

  }
}
